package org.hbhk.hms.mq.tx.consumer;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.hbhk.hms.mq.tx.mq.JsonUtil;
import org.hbhk.hms.mq.tx.mq.MqMsgVo;
import org.hbhk.hms.mq.tx.mq.TypeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ConsumerContainer {

	public static Map<String, DefaultMQPushConsumer> listenerMap = new ConcurrentHashMap<>();
	
	public static Map<String, Integer> listenerThreadNum = new ConcurrentHashMap<>();


	public static Logger log = LoggerFactory.getLogger(ConsumerContainer.class);

	public static Class<?> mqclz = MqMsgVo.class;

	public static <T> void addConsumer(ConsumerConfig config, IMqtxMessageListener<T> messageListener) {

		DefaultMQPushConsumer consumer = null;
		try {
			consumer = new DefaultMQPushConsumer(config.getGroup());
			consumer.setNamesrvAddr(config.getServerUrl());
			consumer.setConsumeThreadMax(config.getThreadMax());
			listenerThreadNum.put(config.getGroup(), config.getThreadMax());
			consumer.setInstanceName(config.getInstanceName());
			consumer.setUnitName(config.getUnitName());
			if (config.getThreadMax() < consumer.getConsumeThreadMin()) {
				consumer.setConsumeThreadMin(config.getThreadMax());
			}
			consumer.subscribe(config.getTopic(), config.getTag());
			Class<?> tclz = TypeUtil.getClassT(messageListener, 0);
			switch (config.getConsumeMode()) {
			case ORDERLY:
				consumer.setMessageListener(new MessageListenerOrderly() {

					@Override
					public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
						return excuteOrder(messageListener, msgs, tclz);
					}
				});
				break;
			case CONCURRENTLY:
				consumer.setMessageListener(new MessageListenerConcurrently() {

					@Override
					public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
							ConsumeConcurrentlyContext context) {
						return excute(messageListener, msgs, tclz);
					}
				});
				break;
			default:
				throw new IllegalArgumentException("property 'consumeMode' was wrong.");
			}

			consumer.start();
		} catch (MQClientException e) {
			log.error("mq start failed", e);
			throw new RuntimeException(e);
		}
		listenerMap.put(config.getGroup(), consumer);
	}

	public static <T> ConsumeOrderlyStatus excuteOrder(IMqtxMessageListener<T> messageListener, List<MessageExt> msgs,
			Class<?> tclz) {
		try {
			MessageExt message = msgs.get(0);
			byte[] json = message.getBody();
			MqMsgVo<T> mqMsgVo = JsonUtil.jsonToObject(json, mqclz, tclz);
			mqMsgVo.put("msgId", message.getMsgId());

			messageListener.onMessage(mqMsgVo);
			return ConsumeOrderlyStatus.SUCCESS;
		} catch (Exception e) {
			log.error("mq consumer error", e);
			return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
		}
	}

	public static <T> ConsumeConcurrentlyStatus excute(IMqtxMessageListener<T> messageListener, List<MessageExt> msgs,
			Class<?> tclz) {
		try {
			MessageExt message = msgs.get(0);
			byte[] json = message.getBody();
			MqMsgVo<T> mqMsgVo = JsonUtil.jsonToObject(json, mqclz, tclz);
			mqMsgVo.put("msgId", message.getMsgId());
			messageListener.onMessage(mqMsgVo);
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		} catch (Exception e) {
			log.error("mq consumer error", e);
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		}
	}

	public static void stop(String config) {
		DefaultMQPushConsumer consumer = listenerMap.get(config);
		if (consumer == null) {
			return;
		}
		consumer.shutdown();
		listenerMap.remove(config);

	}

}
